package BSSource;
import java.io.BufferedReader;

import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;

import BDMSource.BDM;
import BDMSource.BDMData;
import Util.MapUtil;


public class BlockSplit {
	
	public static int partitionIndex;
	public static java.util.HashMap<String, ReduceTask> matchRedTasks = new HashMap<String, ReduceTask>();
	public static ArrayList<ReduceTask> reduceTasks = new ArrayList<ReduceTask>();
	public static int numReduceTasks = 0;
	public static int numMapTasks = 0;
	public static int largestBlockSize = 0;
	
	public static class Map extends Mapper<LongWritable, Text, Text, Text> {
		 
	   private static Text valueTmp = new Text();
	   private static Text word = new Text();
	   public static java.util.Map<String, Double> matchTasks = new HashMap<String, Double>();
	    
	   @Override
	protected void setup(Context context) throws IOException, InterruptedException {
		   getBDM(context);
		   numReduceTasks = Integer.parseInt(context.getConfiguration().get("numReduceTasks"));
		   numMapTasks = Integer.parseInt(context.getConfiguration().get("numMapTasks"));
		   partitionIndex = getMapIndex(context.getInputSplit().toString());
		   getTasks(context);
	   }
	   
	   private void getTasks(Context context) {
	    	matchRedTasks = new HashMap<String, ReduceTask>();
	    	String DirPathMT = context.getConfiguration().get("DirPath") + "matchTasksBS/MatchRedTasks.txt";
	    	try{
	    		FileSystem fs = FileSystem.get(new Configuration());
				BufferedReader buffReader = new BufferedReader(new InputStreamReader(fs.open(new Path(DirPathMT))));
				String line = buffReader.readLine();
				while (line != null){
					String tmpKey = line.substring(0, line.lastIndexOf("\t"));
					StringTokenizer restOfLine = new StringTokenizer(line.substring(line.lastIndexOf("\t")+1, line.length()), ".");
					int reduceTaskIndex = Integer.parseInt(restOfLine.nextToken());
					int comps = Integer.parseInt(restOfLine.nextToken());
					matchRedTasks.put(tmpKey, new ReduceTask(reduceTaskIndex, comps));
					line = buffReader.readLine();
	    		}
			 }catch(Exception e){
				 e.printStackTrace();
			 }
	   }
		
	   public static ReduceTask getNextReduceTask(){
		   ReduceTask minReduce = reduceTasks.get(0);
		   for(int i = 0; i < reduceTasks.size(); i++){
			   if (reduceTasks.get(i).getComps() < minReduce.getComps()){
				   minReduce = reduceTasks.get(i);
			   }
		   }
		   return minReduce;
	   }

	   private void getBDM(Context context) {
		   String DirPathBDM = context.getConfiguration().get("DirPath") + "BDM.txt";
		   BDM.blockingPos = 0;
		   BDM.sumPairs = 0;
		   BDM.BDMData = new ArrayList<BDMData>();
		   BDM.blockingkeyIndexer = new HashMap<String, Integer>();
		   try{
			   FileSystem fs = FileSystem.get(new Configuration());
			   BufferedReader buffReader = new BufferedReader(new InputStreamReader(fs.open(new Path(DirPathBDM))));
			   HashMap<Integer, Integer> partitions = new HashMap<Integer, Integer>();
			   int blockSize = 0;
			   String line = buffReader.readLine();
			   while (line != null){
				   String blockingKey = line.substring(0, 3);
				   int lastDelimiterIndex = line.lastIndexOf("\t");
				   int blockPartSize = Integer.parseInt(line.substring(lastDelimiterIndex+1, line.length()));
				   line = line.substring(0, lastDelimiterIndex);
				   lastDelimiterIndex = line.lastIndexOf(".");
				   int partition = Integer.parseInt(line.substring(lastDelimiterIndex+1, line.length()));
				   partitions.put(partition, blockPartSize);
				   blockSize += blockPartSize;
				   line = buffReader.readLine();
				   if (line == null || !blockingKey.equals(line.substring(0, 3))){
					   if(blockSize > largestBlockSize) largestBlockSize = blockSize;
					   BDM.BDMData.add(new BDMData(blockingKey, partitions, blockSize));
					   BDM.blockingkeyIndexer.put(blockingKey, BDM.blockingPos);
					   BDM.blockingPos++;
					   BDM.sumPairs += (0.5) * blockSize * (blockSize-1);
					   partitions = new HashMap<Integer, Integer>();
					   blockSize = 0;
				   }
				}
				buffReader.close();
		   }catch(Exception e){
			   e.printStackTrace();
		   }
	   }
	   
	   @Override
	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		   String entity = value.toString();
		   String blockingkey = computeKey(entity);
		   int k = BDM.blockIndex(blockingkey);
		   double comps = (0.5)*BDM.size(k)*(BDM.size(k) - 1);
		   double compsPerReduceTask = BDM.pairs()/numReduceTasks;
		   ReduceTask reduceTask = new ReduceTask(0, 0); 
		   if(comps <= compsPerReduceTask){
			   if(comps > 0){
				   reduceTask = matchRedTasks.get("(" + k + "," + 0 + "," + 0 + ")");
				   word.set(reduceTask.getIdReduceTask() + "." + k + ".0.0.0");
				   valueTmp.set("(" + entity + "," + partitionIndex + ")");
				   context.write(word, valueTmp);
			   }
		   }
		   else{
			   for (int i = 0; i < numMapTasks; i++){
				   int min = Math.min(partitionIndex, i);
				   int max = Math.max(partitionIndex, i);
				   reduceTask = matchRedTasks.get("(" + k + "," + max + "," + min + ")");
				   if(reduceTask != null){
					   word.set(reduceTask.getIdReduceTask() + "." + k + "." + max + "." + min);
					   valueTmp.set("(" + entity + "," + partitionIndex + ")");
					   context.write(word, valueTmp);
				   }
			   }
		   }
	   }

	   private int getMapIndex(String taskAttemptID){
		   taskAttemptID = taskAttemptID.substring(0, taskAttemptID.lastIndexOf("."));
		   return Integer.parseInt(taskAttemptID.substring(taskAttemptID.lastIndexOf("_")+1, taskAttemptID.length()));	
	   }
	    
	   private String computeKey(String entity){
		   if (entity.length() < 3) return ">>>";
		   return entity.substring(0, 3).replace(".", ",").toUpperCase();	
	   }
	    
	}
	
	public static void initReduceTasks(int numReduceTasks) {
		for (int i = 0; i < numReduceTasks; i++){
			reduceTasks.add(new ReduceTask(i, 0));
		}
	}
	
	public static class Part extends Partitioner<Text,Text>{
	 	
		@Override
		public int getPartition(Text key, Text value, int numRedTasks) {
			return Integer.parseInt(new StringTokenizer(key.toString(), ".").nextToken());
		}
	 	
	}
	
	public static class NaturalKeyGroupingComparator extends WritableComparator {
	
		protected NaturalKeyGroupingComparator() {
			super(Text.class, true);
		}
		
		@SuppressWarnings("rawtypes")
	    @Override
		public int compare(WritableComparable w1, WritableComparable w2) {
			StringTokenizer stk1 = new StringTokenizer(w1.toString(), ".");
			stk1.nextToken();
			String k1 = stk1.nextToken() + "." + stk1.nextToken() + "." + stk1.nextToken(); 
			StringTokenizer stk2 = new StringTokenizer(w2.toString(), ".");
			stk2.nextToken();
			String k2 = stk2.nextToken() + "." + stk2.nextToken() + "." + stk2.nextToken();
			return k1.compareTo(k2);
		}
	}

	 
	public static class Reduce extends Reducer<Text, Text, Text, Text> {
	
		@Override
		public void reduce(Text key, Iterable<Text> values, Context context) 
			throws IOException, InterruptedException {
	    	
			ArrayList<String> buffer = new ArrayList<String>();
			ArrayList<String> valueList = new ArrayList<String>();
	    	for (Text val: values) {
				valueList.add(val.toString().substring(1, val.toString().length() - 1));
		    }
	    	
			Text entityAux = new Text();	    	
	    	StringTokenizer stAux = new StringTokenizer(key.toString(), ".");
	    	stAux.nextToken();
	    	stAux.nextToken();
	    	int min = Integer.parseInt(stAux.nextToken());
	    	int max = Integer.parseInt(stAux.nextToken());
	    	if(min == max){
	    		for(int i = 0; i < valueList.size(); i++){
	    			String e2 = valueList.get(i);
	    			for(int j = 0; j < buffer.size(); j++){
	    				String e1 = valueList.get(j);
						entityAux.set(match(e1, e2));
						key.set("");
						context.write(key, entityAux);
					}
	    			buffer.add(e2);
	            }
	    	}
	    	else{
	    		String e1 = valueList.get(0);
	    		buffer.add(e1.substring(0, e1.lastIndexOf(",")));
	    		String firstPartitionIndex = e1.substring(e1.lastIndexOf(",")+1, e1.length());
	    		for(int i = 1; i < valueList.size(); i++){
	    			String e2 = valueList.get(i);
		    		String tempPartitionIndex = e2.substring(e2.lastIndexOf(",")+1, e2.length());
		    		if (firstPartitionIndex.equals(tempPartitionIndex)){
		    			buffer.add(e2.substring(0, e2.lastIndexOf(",")));
		    		}
		    		else{
		    			for(int j = 0; j < buffer.size(); j++){
		    				entityAux.set(match(buffer.get(j), e2));
		    				key.set("");
							context.write(key, entityAux);
		    			}
		    		}
	    		}
	    	}
	    }
	    
	    public static String match(String e1, String e2){
			if (e1.length() < 3) e1 = ">>>";
			if (e2.length() < 3) e2 = ">>>";
			//return e1 + "  x  " + e2;
			/*if (new JaroWinkler().score(e1, e2) > 0.6){
				return "Replication: " + e1 + " - " + e2;
			}
			else{*/
			//return e1.substring(0, 3) + e2.substring(0, 3);
			return "-";
			/*}*/
	    }
	}
	  
	public static void map_configure(IntWritable m, IntWritable r) throws IOException, InterruptedException {
		System.out.println("Comparison Size: " + BDM.pairs());
		double compsPerReduceTask = BDM.pairs()/r.get();
    	BDM.readBDM();
    	double comps = 0;
    	
    	for (int k = 0; k <= BDM.numBlocks()-1; k++){
    		comps = (0.5)*BDM.size(k)*(BDM.size(k) - 1);
    		if(comps > 0 && comps <= compsPerReduceTask){
    			Map.matchTasks.put("(" + k + "," + 0 + "," + 0 + ")", comps);
    		}
    		else{
    			if (comps > 0){
    				for (int i = 0; i <= m.get()-1; i++){
    					int FIki = BDM.size(k, i);
    					for (int j = 0; j <= i; j++){
    						int FIkj = BDM.size(k, j);
    						if (FIki * FIkj > 0){
    							if (i == j){
    								Map.matchTasks.put("(" + k + "," + i + "," + j + ")", (0.5) * FIki * (FIki - 1));
    							}
    							else{
    								Map.matchTasks.put("(" + k + "," + i + "," + j + ")", (double) FIki * FIkj);
    							}
    						}
    					}
    				}
    			}
    		}
    	}
    	
    	
    	BlockSplit.Map.matchTasks = MapUtil.sortByValue(Map.matchTasks);
    	Object[] keys = Map.matchTasks.keySet().toArray();
    	
    	BlockSplit.initReduceTasks(r.get());
    	for (int l = keys.length-1; l >= 0; l--){
    		ReduceTask redTask = Map.getNextReduceTask();
    		matchRedTasks.put(keys[l].toString(), redTask);
    		redTask.setComps(redTask.getComps() + Map.matchTasks.get(keys[l]).longValue());
    		reduceTasks.set(redTask.getIdReduceTask(), redTask);
    	}
	}
}

